package parser;

import database.inter.IDB;
import jodis.conn.impl.*;
import jodis.conn.inter.IReply;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ByteChannel;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;

import static tcp.utils.StreamUtils.bufferToBytesArray;
import static tcp.utils.StreamUtils.readStream;

public class Parser {

    public static final int ARRAY = 1;
    public static final int ERROR = 2;
    public static final int NUMBER = 3;
    public static final int STRING = 4;
    public static final int NONE = -1;

//    public ConcurrentLinkedQueue<IReply> linkedQueue = new ConcurrentLinkedQueue<>();

    private JoConnection clientConn;
    private IDB db;

    private static ByteBuffer ret = ByteBuffer.allocate(4096);

    public void setDb(IDB db) {
        this.db = db;
    }


    public void setClientConn(JoConnection clientConn) {
        this.clientConn = clientConn;
    }

    public static ReadState parse(byte[] bs, ReadState state) {
        state.resetSomeStatus();
        while (state.cursor < bs.length) {
            parse0(bs, state);
        }
//        if (state.numOfCmds > state.cmds.size()) {
//            state.cmds.add(state.resultArr);
//        }

        return state;
    }

    private static void parse0(byte[] bs, ReadState state) {
        if (bs[state.cursor] == '*') {
            state.cursor++;
            // 这里 num 代表数组的元素个数
            int num = 0;
            while (state.cursor < bs.length && bs[state.cursor] != '\r') {
                num = num * 10 + bs[state.cursor]-'0';
                state.cursor++;
            }
            state.numOfCmds++;
            state.resultArr = new String[num];
            state.msgType = ARRAY;
        } else if (bs[state.cursor] == '$') {
            state.cursor++;
            // 这里 num 代表字符串的长度
            int num = 0;
            while (state.cursor < bs.length && bs[state.cursor] != '\r') {
                num = num * 10 + bs[state.cursor]-'0';
                state.cursor++;
            }
            // 跳过 \r\n
            state.cursor += 2;

            StringBuffer sb = new StringBuffer();
            for (int i=0;i<num;i++) {
                sb.append((char)bs[state.cursor]);
                state.cursor++;
            }
            if (state.msgType != ARRAY) {
                state.msgType = STRING;
                state.resultStr = sb.toString();
            } else {
                state.resultArr[state.count++] = sb.toString();
                if (state.count == state.resultArr.length) {
                    state.cmds.add(state.resultArr);
                    state.count = 0;
                }
            }

        } else if (bs[state.cursor] == '+' || bs[state.cursor] == '-' || bs[state.cursor] == ':') {
            System.out.println((char) bs[state.cursor]);
            // 简单字符串，后面直接就是字符串，不会再加上字符串的长度
            state.cursor++;
            StringBuffer sb = new StringBuffer();
            while (state.cursor < bs.length && bs[state.cursor] != '\r') {
                sb.append((char)bs[state.cursor]);
                state.cursor++;
            }

            if (state.msgType != ARRAY) {
                switch (bs[0]) {
                    case '+':
                        state.msgType = STRING;
                        break;
                    case '-':
                        state.msgType = ERROR;
                        break;
                    case ':':
                        state.msgType = NUMBER;
                        break;
                }
                state.resultStr = sb.toString();
            } else {
                state.resultArr[state.count++] = sb.toString();
            }
        } else if (isCharacterOrNumber(bs[state.cursor])) {
            StringBuffer sb = new StringBuffer();
            while (state.cursor < bs.length && bs[state.cursor] != '\r') {
                sb.append((char)bs[state.cursor]);
                state.cursor++;
            }
            state.cmds.add(new String[]{sb.toString()});
        } else {
            state.msgType = NONE;
        }

        // 处理完后，跳过 \r\n
        state.cursor+=2;
    }

    public static boolean isCharacterOrNumber(byte c) {
        return (c >= '0' && c <= '9') ||
                (c >= 'a' && c <= 'z') ||
                (c >= 'A' && c <= 'Z');
    }

    public void parse1(ByteChannel channel, ByteBuffer buf, ReadStateNIO state) throws IOException {
        while (true) {
            byte[] msg = readLine(channel, buf, state);
            if (msg == null) {
                continue;
            }
            parseWithNIO(msg, state);
        }
    }

    public void parseWithNIO(byte[] msg, ReadStateNIO state) throws IOException {
        if (msg == null) {
            return;
        }
        // 有可能 buf 读完了，但是还没读到 '\n', 那么需要继续读数据
        if (!state.isEnough) {
            return;
        }
        if (!state.readMultiLine) {
            int type = msg[0] & 0xff;
            if (type == '*') {
                boolean res = parseMultiBulkHeader(msg, state);
                if (!res) {
                    state.reset();
                    return;
                }

                if (state.count == 0) {
                    state.reset();
//                    linkedQueue.add(EmptyMultiBulkReply.makeEmptyMultiBulkReply());
                }
            } else if (type == '$' || type == '#') {
                boolean res = parseBulkHeader(msg, state);
                if (!res) {
                    state.reset();
                }
            } else {
                // 简单字符串
                IReply reply = parseSingleLine(msg);
                if (reply instanceof MultiBulkReply) {
                    MultiBulkReply mPayload = (MultiBulkReply) reply;
                    IReply r = db.exec(clientConn, mPayload.getArgs());
                    clientConn.writeWithNIO(r.toBytes());
                }
//                linkedQueue.add(reply);
            }
        } else {
            // 按照更新后的 state 字段，来解析
            readBody(msg, state);
            if (state.finish()) {
                if (state.type == '*') {
                    MultiBulkReply payload = MultiBulkReply.makeMultiBulkReply(state.args);
                    IReply reply = db.exec(clientConn, payload.getArgs());
                    clientConn.writeWithNIO(reply.toBytes());
//                    linkedQueue.add(payload);
                } else if (state.type == '$') {
                    MultiBulkReply payload = MultiBulkReply.makeMultiBulkReply(state.args);
                    IReply reply = db.exec(clientConn, payload.getArgs());
                    clientConn.writeWithNIO(reply.toBytes());
//                    linkedQueue.add(payload);
                }
                state.reset();
            }
        }

    }

    public static void readBody(byte[] msg, ReadStateNIO state) {
        if (msg.length > 0 && (msg[0] & 0xff) == '$') {
            state.bulkLen = Integer.parseInt(new String(msg).substring(1, msg.length-2));
        } else {
            // 把 \r\n 都加进去了
            state.args[state.index++] = Arrays.copyOfRange(msg, 0, msg.length);
        }

    }

    public static IReply parseSingleLine(byte[] msg) {
        String ss = new String(msg).trim();
        int code = msg[0] & 0xff;
        switch (code) {
            case '+':
                // 简单字符串
                return StatusReply.MakeStatusReply(ss);
            case '-':
                // 错误信息
                return StandardErrReply.makeErrReply(ss);
            case ':':
                // 整数
                int num = Integer.parseInt(ss);
                return IntReply.makeIntReply(num);
            default:
                // 直接当文本解析
                String[] strs = ss.split("\\s");
                return MultiBulkReply.makeMultiBulkReply(strs);
        }
    }

    public static boolean parseBulkHeader(byte[] msg, ReadStateNIO state) {
        // 获取字符串长度
        int num = Integer.parseInt(new String(msg).substring(1, msg.length-2));
        if (num > 0) {
            state.type = msg[0] & 0xff;
            state.readMultiLine = true;
            state.count = 1;
            state.args = new byte[1][];
            return true;
        }
        return false;
    }

    public static boolean parseMultiBulkHeader(byte[] msg, ReadStateNIO state) {
        // 获取元素个数
        int num = Integer.parseInt(new String(msg).substring(1, msg.length-2));
        if (num == 0) {
            state.count = 0;
            return true;
        } else if (num > 0) {
            state.type = msg[0] & 0xff;
            state.count = num;
            state.readMultiLine = true;
            state.args = new byte[num][];
            return true;
        }
        return false;

    }

    public static byte[] readLine(ByteChannel channel, ByteBuffer buf, ReadStateNIO state) throws IOException {
//        ByteBuffer ret = ByteBuffer.wrap(new byte[4096]);
//        ByteBuffer ret = ByteBuffer.allocate(4096);
        if (buf.position() == 0 || !buf.hasRemaining()) {
            buf.clear();
            int r = channel.read(buf);
            if (r == 0 || r == -1) {
                state.reachEOF = r == -1;
                return null;
            }
            buf.flip();
        }

        if (state.bulkLen == 0) {
            while (buf.hasRemaining()) {
                // 通过 get，让 position 不断前移
                byte c = buf.get();
                ret.put(c);
                if (c == (byte)'\n') {
                    state.isEnough = true;
                    break;
                }
            }
            // 有可能 buf 读完了，但是还没读到 '\n'
        } else {
            // +2 是为了 \r\n
            int bulkLen = state.bulkLen + 2;
            for (int i=0;i<bulkLen;i++) {
                if (buf.hasRemaining()) {
                    // 通过 get，让 position 不断前移
                    ret.put(buf.get());
                }
            }
            state.bulkLen = 0;
        }
        // It sets the limit to the current position and then sets the position to zero.
        ret.flip();
        byte[] bs = new byte[ret.limit()];
        ret.get(bs);
//        byte[] bs = bufferToBytesArray(ret);
        // It sets the limit to the capacity and the position to zero.
        ret.clear();
        return bs;
    }

    public static void main(String[] args) {
        try {
//            BufferedInputStream in = new BufferedInputStream(new FileInputStream("redis_benchmark.bin"));
            File f = new File("redis_benchmark.bin");
            FileChannel fc = FileChannel.open(f.toPath());
            ByteBuffer bb = ByteBuffer.wrap(new byte[4096]);
            ReadStateNIO state = ReadStateNIO.getDefaultState();
            Parser parser = new Parser();

            parser.parse1(fc, bb, state);


//            while (!parser.linkedQueue.isEmpty()) {
//                IReply reply = parser.linkedQueue.poll();
//                System.out.println(new String(reply.toBytes()));
//            }

            fc.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

